package com.rxjava;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;

public class SimpleDemo2 {

	static String str = "";

	public Observable<String> send() {
		Observable<String> observable = Observable.just("just1", "just2");//依次发送"just1"和"just2"
		return observable;
	}

	public Observable<String> send2() {
		List<String> list = new ArrayList<>();
		list.add("from1");
		list.add("from2");
		list.add("from3");
		/**
		 * 注意，just()方法也可以传list，但是发送的是整个list对象，而fromIterable（）发送的是list的一个item
		 */
		Observable<String> observable = Observable.fromIterable(list);  //遍历list 每次发送一个
		return observable;
	}

	public Observable<Integer> send3() {
		//使用range( ),创建一个发射特定整数序列的Observable，第一个参数为起始值，第二个为发送的个数，如果为0则不发送，负数则抛异常：
		Observable<Integer> observable = Observable.range(10, 5);
		return observable;
	}

	public Observable<Long> send4() {
		Observable<Long> observable = Observable.timer(3, TimeUnit.SECONDS);  //3秒后发射一个值
		//Observable<Long> observable = Observable.interval(3, TimeUnit.SECONDS);  //3秒1次发射一个值
		return observable;
	}

	public Observable<String> send5() {
		Observable<String> observable = Observable.just("repeatObservable").repeat(3);//重复发射3次(当.repeat()接收到.onCompleted()事件后触发重订阅)
		return observable;
	}

	public Observable<String> send6() {
		//当观察者订阅时，才创建Observable，并且针对每个观察者创建都是一个新的Observable。以何种方式创建这个Observable对象，当满足回调条件后，就会进行相应的回调。
		Observable<String> observable = Observable.defer(new Callable<ObservableSource<? extends String>>() {
            @Override
            public ObservableSource<? extends String> call() throws Exception {
                return Observable.just("hello");
            }
        });
		return observable;
	}

	public Observable<String> send7() {
		//当观察者订阅时，才创建Observable，并且针对每个观察者创建都是一个新的Observable。以何种方式创建这个Observable对象，当满足回调条件后，就会进行相应的回调。
		Observable<String> observable = Observable.defer(new Callable<ObservableSource<? extends String>>() {
			@Override
			public ObservableSource<? extends String> call() throws Exception {
				return Observable.just(str);
			}
		});
		return observable;
	}

	public Observer<String> revieve() {
		// 创建一个下游 Observer
		Observer<String> observer = new Observer<String>() {
			@Override
			public void onSubscribe(Disposable d) {
				System.out.println("subscribe");
			}

			@Override
			public void onNext(String value) {
				System.out.println(value);
			}

			@Override
			public void onError(Throwable e) {
				System.out.println("error");
			}

			@Override
			public void onComplete() {
				System.out.println("complete");
			}
		};
		return observer;
	}

	public Observer<Integer> revieve2() {
		// 创建一个下游 Observer
		Observer<Integer> observer = new Observer<Integer>() {
			@Override
			public void onSubscribe(Disposable d) {
				System.out.println("subscribe");
			}

			@Override
			public void onNext(Integer value) {
				System.out.println(value);
			}

			@Override
			public void onError(Throwable e) {
				System.out.println("error");
			}

			@Override
			public void onComplete() {
				System.out.println("complete");
			}
		};
		return observer;
	}

	public Observer<Long> revieve3() {
		// 创建一个下游 Observer
		Observer<Long> observer = new Observer<Long>() {
			@Override
			public void onSubscribe(Disposable d) {
				System.out.println("subscribe");
			}

			@Override
			public void onNext(Long value) {
				System.out.println(value);
			}

			@Override
			public void onError(Throwable e) {
				System.out.println("error");
			}

			@Override
			public void onComplete() {
				System.out.println("complete");
			}
		};
		return observer;
	}

	//Empty：创建一个空的，但可以正常结束的事件源
	//Never：创建一个空的，永远不会结束的事件源
	//Throw：创建一个空的，会抛出异常的事件源

	public static void main(String[] args) {
		SimpleDemo2 demo = new SimpleDemo2();
		demo.send().subscribe(demo.revieve());
		System.out.println("===================================1");
		demo.send2().subscribe(demo.revieve());
		System.out.println("===================================2");
		demo.send3().subscribe(demo.revieve2());
		System.out.println("===================================3");
		demo.send4().subscribe(demo.revieve3());
		System.out.println("===================================4");
		demo.send5().subscribe(demo.revieve());
		System.out.println("===================================5");
		demo.send6().subscribe(demo.revieve());
		System.out.println("===================================6");
		Observable<String> observable = demo.send7();
		//defer是在订阅的时候才会赋值，所以可以在这里赋值
		str = "aaa";
		observable.subscribe(demo.revieve());
	}

}
